package com.hjd.shop.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.mapper.Condition;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import com.hjd.shop.api.service.SOrderPayService;
import com.hjd.shop.base.Response;
import com.hjd.shop.constant.PayState;
import com.hjd.shop.constant.ShopCode;
import com.hjd.shop.constant.SystemUser;
import com.hjd.shop.entity.MqProducerLog;
import com.hjd.shop.entity.SOrderPay;
import com.hjd.shop.exception.CastException;
import com.hjd.shop.mapper.MqProducerLogMapper;
import com.hjd.shop.mapper.SOrderPayMapper;
import com.hjd.shop.utils.ThreadPool;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @Author: hjd
 * @Date: 2020/11/1 0001 15:53
 */
@Component
@Service(interfaceClass = SOrderPayService.class)
@Slf4j
public class SOrderPayServiceImpl extends ServiceImpl<SOrderPayMapper, SOrderPay> implements SOrderPayService {
    @Autowired
    private SOrderPayMapper orderPayMapper;
    @Autowired
    private MqProducerLogMapper mqProducerLogMapper;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private ThreadPool threadPool;

    @Value("${rocketmq.producer.group}")
    private String groupName;
    @Value("${mq.pay.topic}")
    private String topic;
    @Value("${mq.pay.tag}")
    private String tag;

    @Override
    public Response createOrderPay(SOrderPay orderPay) {
        if (orderPay.getOrderId()==null) {
            CastException.cast(ShopCode.ARGS_IS_NOT_EXIT);
        }
        //1.判断订单支付状态
        Condition condition = Condition.create();
        condition.eq("order_id",orderPay.getOrderId())
                .eq("is_paid", PayState.ORDER_PAYED);
        int count = this.selectCount(condition);
        if(count>0){
            CastException.cast(ShopCode.ORDER_IS_PAYED);
        }
        //2.将订单状态改为已支付
        orderPay.setIsPaid(PayState.ORDER_UN_PAY.getCode());
        //3.保存订单
        orderPayMapper.insert(orderPay);
        //4.返回结果
        return new Response(ShopCode.SHOP_SUCCESS.getMsg());
    }

    @Override
    public Response payedCallBack(SOrderPay orderPay)  {
        log.info("开始执行支付回调...");
        /**
         * 将 SOrderPay 转换为字符串作为消息体
         */
        String body= JSON.toJSONString(orderPay);
        /**
         * 以订单Id作为业务Key
         */
        String key=orderPay.getOrderId();
        //1.判断用户支付的状态
        if (orderPay.getIsPaid()==(PayState.PAY_STATION_SUCCESS.getCode())) {
            //2.更新支付订单状态为已支付
            orderPay.setIsPaid(PayState.ORDER_PAYED.getCode())
                    .setModifyTime(new Date())
                    .setModifyUser(SystemUser.SYSTEM_USER);
            Integer result = orderPayMapper.updateById(orderPay);
            if (result==1) {
                log.info("更新订单成功!");
                //3.创建 支付成功的 mq消息
                log.info("开始参加mq消息...");
                MqProducerLog mqProducerLog = new MqProducerLog();
                mqProducerLog.setGroupName(groupName)
                        .setMsgTopic(topic)
                        .setMsgTag(tag)
                        .setMsgKey(key)
                        .setMsgBody(body)
                        .setCreateTime(new Date())
                        .setCreateUser(SystemUser.SYSTEM_USER);
                //4.为了保证mq消息发送成功，需要将消息持久化
                mqProducerLogMapper.insert(mqProducerLog);
                log.info("订单消息持久化成功!");
                //5.发送消息到MQ
                /**
                 * 这里 发送消息后 还需要等待 mq的返回，由于是在主线程中进行执行的，
                 * 当支付的消息过多时，会导致消息在这里堆积，效率很差很慢。解决方案就是
                 * 在线程池中进行处理
                 */
                threadPool.getThreadPool().submit(new Runnable() {
                    @Override
                    public void run() {
                        log.info("开始发送消息...");
                        SendResult sendResult= null;
                        try {
                            sendResult = sendMessage(tag,topic,key,body);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        log.info("发送消息成功!");
                        //6.等待发送的结果,如果mq接收到消息，删除mq消息在数据库
                        if (sendResult.equals(SendStatus.SEND_OK)) {
                            log.info("消息返回成功!");
                            mqProducerLogMapper.deleteById(mqProducerLog);
                        }
                    }
                });
            }
            return new Response(ShopCode.SHOP_SUCCESS.getMsg());
        }else {
            CastException.cast(false,"订单:"+orderPay.getOrderId()+"状态不是【已支付】,在支付回调过程中!");
            return new Response(ShopCode.SHOP_FILED.getMsg());
        }
    }

    /**
     * 向mq发送支付成功消息
     * @param tag
     * @param topic
     * @param key
     * @param body
     */
    private SendResult sendMessage(String tag, String topic, String key, String body) throws InterruptedException,
            RemotingException, MQClientException, MQBrokerException {
        if (StringUtils.isEmpty(topic)) {
            CastException.cast(false,"向mq发送支付成功消息时，topic为空!");
        }
        if (StringUtils.isEmpty(tag)) {
            CastException.cast(false,"向mq发送支付成功消息时，tag为空!");
        }
        Message message = new Message(topic,tag,key,body.getBytes());
        SendResult send = rocketMQTemplate.getProducer().send(message);
        return send;
    }
}
